package cn.edu.gdut.deepcode.task;

import cn.edu.gdut.deepcode.component.GlobalIdGenerator;
import cn.edu.gdut.deepcode.component.MinIOService;
import cn.edu.gdut.deepcode.component.RedisStreamService;
import cn.edu.gdut.deepcode.config.SnapshotTaskProperties;
import cn.edu.gdut.deepcode.domain.po.CanvasSnapshot;
import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.extension.toolkit.Db;
import lombok.Setter;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;

import java.io.BufferedInputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import static cn.edu.gdut.deepcode.constant.AbnormalInfoConstant.CST_EI_1;
import static cn.edu.gdut.deepcode.constant.CacheConstant.*;

/**
 * <p>
 * 快照任务管理器
 * </p>
 *
 * @Author: 暮倦
 * @Since: 2025-04-13  10:13
 * @Version: 1.0.0
 */

@Slf4j
public class SnapshotTaskManager extends Thread {
    // stream流服务
    private final RedisStreamService redisStreamService;
    // minio服务
    private final MinIOService minIOService;
    // 非核心线程启动的时间
    private Long startTime = null;
    // 永久的存活时间
//    private final Long permanentLifeTime = -1L;
    // 快照任务管理器的任务队列
    private final LinkedBlockingDeque<Long> managerTaskQueue;
    // 快照生成任务线程池
    private final ExecutorService snapshotGenerationTaskExecutor;
    // 快照生成任务消费者集合
    private final ArrayList<SnapshotGenerationTaskConsumer> snapshotGenerationTaskConsumers;
    // 快照任务相关配置
    private final SnapshotTaskProperties properties;
    // 快照生成任务id生成器
    private final GlobalIdGenerator globalIdGenerator;
    // 画布id->消费者索引映射表
    private final Map<Long, Integer> hashMap = new HashMap<>();

    public SnapshotTaskManager(ExecutorService snapshotGenerationTaskExecutor,
                               SnapshotTaskProperties properties,
                               GlobalIdGenerator globalIdGenerator,
                               RedisStreamService redisStreamService,
                               MinIOService minIOService
    ) {
        // 为容器赋值
        this.managerTaskQueue = new LinkedBlockingDeque<>(properties.getManagerTaskQueueSize());
        this.snapshotGenerationTaskExecutor = snapshotGenerationTaskExecutor;
        this.snapshotGenerationTaskConsumers = new ArrayList<>();
        // 存储配置,设置id生成器,设置stream服务、minio服务
        this.properties = properties;
        this.globalIdGenerator = globalIdGenerator;
        this.redisStreamService = redisStreamService;
        this.minIOService = minIOService;
        // 初始化快照生成任务消费者、线程池
        initGenSnapshotTaskConsumer();
    }

    private void initGenSnapshotTaskConsumer() {
        for (int i = 0; i < properties.getCore(); i++) {
            SnapshotGenerationTaskConsumer snapshotGenerationTaskConsumer = genPermanent();
            snapshotGenerationTaskConsumers.add(snapshotGenerationTaskConsumer);
            snapshotGenerationTaskExecutor.submit(snapshotGenerationTaskConsumer);
        }
    }

    private SnapshotGenerationTaskConsumer gen(Long lifeTime) {
        return new SnapshotGenerationTaskConsumer(
                globalIdGenerator.generate(GLOBAL_ID_SNAPSHOT_GENERATION_TASK_CONSUMER_KEY),
                new LinkedBlockingQueue<>(properties.getSnapshotGenerationTaskQueueSize()),
                lifeTime,
                properties.getTimeUnit()
        );
    }

    private SnapshotGenerationTaskConsumer genPermanent() {
        return gen(PERMANENT);// 永久的存活时间
    }

    private SnapshotGenerationTaskConsumer genNonPermanent() {
        return gen(properties.getLifeTime());
    }

    private int genConsumerIdx(Long canvasId) {
        return (int) (canvasId % snapshotGenerationTaskConsumers.size());
    }


    @Override
    public void run() {
        Integer consumerIdx = null;
        // 根据映射表挑选消费者是否成功的标识,false表示挑选失败,true表示挑选成功
        boolean flag = true;
        while (true) {
            // 取出画布id
            Long canvasId = managerTaskQueue.poll();
            // 如果有必要的话新增消费者
            createConsumerIfNecessary(canvasId);
            // 挑选消费者,先根据映射表进行挑选
            consumerIdx = hashMap.get(canvasId);
            // 如果挑选失败(consumerIdx为null)则根据hash运算,为当前快照任务选择合适的快照生成任务消费者索引
            if (consumerIdx == null || snapshotGenerationTaskConsumers.size() <= consumerIdx) {
                flag = false;
                consumerIdx = genConsumerIdx(canvasId);
            }
            // 获取该位置的快照生成任务消费者
            SnapshotGenerationTaskConsumer snapshotGenerationTaskConsumer = snapshotGenerationTaskConsumers.get(consumerIdx);
            if (!snapshotGenerationTaskConsumer.offer(canvasId)) {
                // 快照生成任务消费者已被终结则创建一个快照生成任务消费者
                snapshotGenerationTaskConsumer = genNonPermanent();
                snapshotGenerationTaskConsumers.set(consumerIdx, snapshotGenerationTaskConsumer);
                // 推送快照生成任务给新的快照生成任务消费者
                snapshotGenerationTaskConsumer.offer(canvasId);
                // 启动消费者
                snapshotGenerationTaskExecutor.submit(snapshotGenerationTaskConsumer);
            }
            if (!flag) {
                // 映射表中不存在映射或映射信息过期则新增或更新记录到映射表
                hashMap.put(canvasId, consumerIdx);
            }
            // 如果有必要则清理消费者
            clearConsumerIfNecessary();
        }
    }

    private void createConsumerIfNecessary(Long canvasId) {
        // 查看当前快照生成任务消费者的数量,数量不大于maxSize则创建新的快照生成任务消费者(设置存活时间为lifeTime)
        if (snapshotGenerationTaskConsumers.size() <= properties.getMaxSize()

        ) {
            SnapshotGenerationTaskConsumer snapshotGenerationTaskConsumer = genNonPermanent();
            snapshotGenerationTaskConsumers.add(snapshotGenerationTaskConsumer);
            snapshotGenerationTaskExecutor.submit(snapshotGenerationTaskConsumer);
        }
    }

    private void clearConsumerIfNecessary() {
        if (snapshotGenerationTaskConsumers.size() <= properties.getCore()) {
            // 无限时消费者则不用尝试清空消费者
            return;
        }
        long curTime = System.currentTimeMillis();
        if (startTime == null) {
            // 有限时消费者但是没有设置startTime
            startTime = curTime;
            return;
        }
        if (curTime - startTime <= properties.getTimeUnit().toMillis(properties.getLifeTime())) {
            // 大概率还存活
            return;
        }
        // 清除非存活消费者
        snapshotGenerationTaskConsumers.removeIf(consumer -> consumer.finished());
        if (properties.getCore() < snapshotGenerationTaskConsumers.size()) {
            // 清除后还有限时消费者则设置startTime为当前时间
            startTime = System.currentTimeMillis();
        } else {
            // 没有则设置为null
            startTime = null;
        }
    }


    /**
     * <p>
     * 快照任务
     * </p>
     *
     * @Author: 暮倦
     * @Since: 2025-04-12  23:16
     * @Version: 1.0.0
     */
    @ToString
    private class SnapshotGenerationTaskConsumer implements Runnable {

        private final Long consumerId;

        private final LinkedBlockingQueue<Long> snapshotGenerationTaskQueue;

        private volatile Long lifeTime;

        private final TimeUnit timeUnit;

        public SnapshotGenerationTaskConsumer(
                Long consumerId,
                LinkedBlockingQueue<Long> snapshotGenerationTaskQueue,
                Long lifeTime,
                TimeUnit timeUnit
        ){
            this.consumerId=consumerId;
            this.snapshotGenerationTaskQueue=snapshotGenerationTaskQueue;
            this.timeUnit=timeUnit;
            this.lifeTime=lifeTime;
        }

        @Setter
        private volatile boolean status = false;

        public boolean finished() {
            return this.status;
        }

        public int getSize() {
            return snapshotGenerationTaskQueue.size();
        }

        public boolean offer(Long canvasId) {
            if (finished()){
                return false;
            }
            snapshotGenerationTaskQueue.offer(canvasId);
            return true;
        }

        @Override
        public void run() {
            while (true) {
                Long canvasId;
                // 发现消费者已被终结且队列大小为空则直接退出循环
                if (finished() && snapshotGenerationTaskQueue.isEmpty()) {
                    return;
                }
                // 从阻塞队列中获取画布id
                if (PERMANENT.equals(lifeTime)) {
                    canvasId = snapshotGenerationTaskQueue.poll();
                } else {
                    // 存活时间为限时则调用限时阻塞方法
                    try {
                        canvasId = snapshotGenerationTaskQueue.poll(lifeTime, timeUnit);
                    } catch (InterruptedException e) {
                        log.debug(CST_EI_1, List.of(this));
                        continue;
//                        CommonUtils.recordErrorInfoAndThrowException(
//                                CST_EI_2,
//                                List.of(this)
//                        );
                    }
                }
                // 如果取出的canvasId为null,说明poll超时了,设置状态为完成状态
                if (canvasId == null) {
                    setStatus(true);
                    // 再次检查队列大小,如果不为零则再次进入循环,如果为零则终结消费者
                    if (snapshotGenerationTaskQueue.isEmpty()) {
                        return;
                    } else {
                        continue;
                    }
                }
                // 设置当前画布快照任务正在进行
                redisStreamService.getRedisTemplate()
                        .opsForValue().set(
                                CANVAS_SNAPSHOT_GENERATION_TASK_STATUS_KEY_PREFIX + canvasId,
                                CANVAS_SNAPSHOT_GENERATION_TASK_STARTING_STATUS,
                                CANVAS_SNAPSHOT_GENERATION_TASK_STATUS_KEY_EXPIRE_TIME,
                                CANVAS_SNAPSHOT_GENERATION_TASK_STATUS_KEY_EXPIRE_TIME_UNIT
                        );
                try {
                    // 从redis中获取当前画布快照的版本号加一后的值
                    Long newVersion = redisStreamService.getRedisTemplate()
                            .opsForValue().increment(CANVAS_SNAPSHOT_VERSION_KEY_PREFIX + canvasId);
                    if (newVersion == null) {
                        // 版本号异常
                        log.warn(CST_EI_1, List.of(canvasId, newVersion));
                        continue;
                    }
                    // 根据画布id,画布快照version从mysql中的快照表获取画布快照
                    long version = newVersion - 1;
                    CanvasSnapshot snapshot = null;
                    if (version > 0) {
                        // 版本号大于0才有必要查询
                        snapshot = Db.lambdaQuery(CanvasSnapshot.class)
                                .eq(CanvasSnapshot::getCanvasId, canvasId)
                                .eq(CanvasSnapshot::getVersion, version)
                                .one();
                        if (BeanUtil.isEmpty(snapshot)) {
                            // 快照信息异常
                            log.warn(CST_EI_1, List.of(canvasId, newVersion, snapshot));
                            continue;
                        }
                    }
                    // 根据画布id、画布快照的最新版本号从minio获取快照数据
                    byte[] curDocByteArr = null;
                    if (snapshot != null) {
                        // 有快照才下载
                        BufferedInputStream stream =
                                new BufferedInputStream(
                                        minIOService.download(canvasId + "/" + version));
                        curDocByteArr = stream.readAllBytes();
                        if (curDocByteArr == null || curDocByteArr.length == 0) {
                            log.warn(CST_EI_1, List.of(canvasId, version, snapshot, curDocByteArr));
                        }
                    }
                    // 从redis的streams流中获取最新未消费的前1000条该画布的操作日志
                    List<Map<Object, Object>> data = redisStreamService.pollDataFromStream(
                            CANVAS_SNAPSHOT_GENERATION_TASK_STREAM_KEY_PREFIX,
                            CANVAS_SNAPSHOT_GENERATION_TASK_GROUP_NAME_PREFIX,
                            CANVAS_SNAPSHOT_GENERATION_TASK_CONSUMER_NAME_PREFIX,
                            CANVAS_SNAPSHOT_GENERATION_TASK_DATA_BATCH_SIZE
                    );
                    if (data==null|| data.isEmpty()){
                        // stream流中无数据,直接跳过
                        log.error(CST_EI_1,List.of(canvasId, version, snapshot, curDocByteArr,data));
                        continue;
                    }
                    // 循环操作日志将所有操作日志和快照数据合并起来
                    Long offset=null;
                    // 如果抛出异常则退出循环

                    // 确认已消费消息

                    // 更新mysql快照信息表
                    Db.save(
                            new CanvasSnapshot()
                                    .setCanvasId(canvasId)
                                    .setVersion(version)
                                    .setOffset(offset)
                    );
                    // 将新的快照存入minio
                    minIOService.upload(null,-1,-1,null);
                } catch (Exception e) {
                    log.error(CST_EI_1, canvasId, e);
                } finally {
                    // 设置当前快照生成任务已结束
                    redisStreamService.getRedisTemplate()
                            .delete(CANVAS_SNAPSHOT_GENERATION_TASK_STATUS_KEY_PREFIX + canvasId);
                }
            }
        }
    }


}


